package sparkStream

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import java.util.Properties

object SparkKafka {
  def main(args: Array[String]): Unit = {
    // 1） 构建sparkconf 本地运行，运行应用程序名称
    val conf = new SparkConf().setMaster("local[*]").setAppName("MINA")

    // 2） 构建 sparkstreaming ---> streamingContext, 加载配置
    // StreamingContext 需要导入依赖
    // spark streaming 可以进行流式处理，微批次处理， 间隔2秒
    val ssc = new StreamingContext(conf, Seconds(2))

    // SPARK 输出红色 info信息  --》 error
    ssc.sparkContext.setLogLevel("error")

    // 3） kafka 配置 broker , key value, group id,消费模式
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "123.56.187.176:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "niit",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    // 4) SPARK 链接 KAFKA 订阅，TOPIC, STREAMINGCONTEXT
    // topic name
    val topicName = Array("stuInfo")
    val streamRdd = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topicName, kafkaParams)
    )

    // 返回KAFKA 返回的 STREAMRDD (一段有时间间隔的RDD）
    val res = streamRdd.map(_.value())
    val result = res.flatMap(_.split(" ")).map((_, 1)).reduceByKeyAndWindow((a: Int, b: Int) => a + b, Seconds(4), Seconds(4))

    // 定义 KafkaProducer 的配置
    val property = new Properties()
    property.put("bootstrap.servers", "node1:9092")
    property.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    property.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    result.foreachRDD { rdd =>
      rdd.foreachPartition { partition =>
        // 创建新的客户端
        val producer = new KafkaProducer[String, String](property)
        try {
          partition.foreach { obj =>
            println("--------数据是------")
            println(obj)
            // 将单词和出现次数格式化为键值对字符串
            val keyValuePair = s"${obj._1}:${obj._2}"
            // 发送消息到 Kafka
            producer.send(new ProducerRecord[String, String]("15Homework", obj._1, keyValuePair))
          }
        } catch {
          case e: Exception => e.printStackTrace()
        } finally {
          // 关闭
          producer.close()
        }
      }
    }

    // 6) 开启SSC , 监控 KAFKA 数据
    ssc.start()
    ssc.awaitTermination()  // 监控KAFKA的数据
  }
}